package plgbase

import (
	"dgo/goutils/utils"
	"fmt"
	"gitee.com/ymofen/gobase/golog"
	"gitee.com/ymofen/gobase/gotask/chantask"
	"gitee.com/ymofen/gobase/subpub"
	"gitee.com/ymofen/supervalue"
	"io"
	"reflect"
	"strings"
	"sync/atomic"
	"time"
)

/*
**

	input
*/
type InnerEx struct {
	releaseFlag int32

	pubque     string
	pubqueTask *chantask.ChanTask
	pubfun     func(topic string, args ...interface{}) int

	ownerPlug      interface{}
	objdna         uint32
	TypeStr        string
	plug           interface{}
	plugID         string
	subscribeTopic string
	subid          string

	topic_prefix     string
	task_topic       string
	task_interval    int
	task_intervalDNA int
	templ            string // json 默认模版
	jsonfields       string // 如果设置则进行copy
	jsonTempl        *wrapper.SuperValue
	confIsChanged    bool // 最后一次是否改变
	TagInt           int
}

func NewInnerEx() *InnerEx {
	return &InnerEx{}
}

func NewInnerEx2(owner interface{}) *InnerEx {
	rval := &InnerEx{}
	rval.ConfigOwnerPlug(owner)
	return rval
}

func (this *InnerEx) ReleaseFlag() int32 {
	return atomic.LoadInt32(&this.releaseFlag)
}

func (this *InnerEx) IsValid() bool {
	if atomic.LoadInt32(&this.releaseFlag) == 1 {
		return false
	}
	ok := this.plug != nil
	if !ok && len(this.plugID) > 0 {
		ok = this.CheckGetPlug() != nil
	}
	return ok
}

func (this *InnerEx) ConfIsChanged() bool {
	return this.confIsChanged
}

func (this *InnerEx) GetInnerPlugId() string {
	return this.plugID
}

func (this *InnerEx) ConfigOwnerPlug(plg interface{}) {
	this.ownerPlug = plg
}

func (this *InnerEx) ConfigJsonFields(v string) {
	this.jsonfields = v
}

func (this *InnerEx) Reset() {
	this.checkClosePubQue()
	this.UnSubscribePre()
	this.plug = nil
	this.plugID = ""
}

func (this *InnerEx) SetTaskTopic(topic string) {
	if this.task_topic != topic {
		this.task_topic = topic
	}
}

func (this *InnerEx) SetInnerPlugId(id string) {
	if this.plugID != id {
		this.UnSubscribePre()
		this.plugID = id
		this.plug = nil
	}
}

func (this *InnerEx) Release() {
	if atomic.CompareAndSwapInt32(&this.releaseFlag, 0, 1) {
		this.UnSubscribePre()
		this.plug = nil
		this.ownerPlug = nil
		this.checkClosePubQue()
	}
}

func (this *InnerEx) GetTempl() string {
	rval := ReplacePlaceholderForFileName(this.templ)
	rval = strings.ReplaceAll(rval, "$plugid$", GetPlugID(this.ownerPlug, ""))
	rval = strings.ReplaceAll(rval, "$now$", utils.NowString())
	return rval
}

func (this *InnerEx) CreateReportJSONCache(usecache bool) *wrapper.SuperValue {
	if len(this.templ) > 0 {
		json, _ := wrapper.NewSVFromBufCache([]byte(this.GetTempl()), usecache)
		if json != nil {
			return json
		}
	}
	return wrapper.NewSVObjectCache(usecache)
}

func (this *InnerEx) CreateReportJSON() *wrapper.SuperValue {
	return this.createReportJSON()
}

func (this *InnerEx) FillTempl(json *wrapper.SuperValue) {
	if len(this.templ) > 0 {
		templ, _ := wrapper.NewSVFromBufCache([]byte(this.GetTempl()), true)
		if templ != nil {
			defer templ.Clear()
			json.CopyFromCache(templ, true)
		}
	}
}

func (this *InnerEx) createReportJSON() *wrapper.SuperValue {
	if len(this.templ) > 0 {
		json, _ := wrapper.NewSVFromBuf([]byte(this.GetTempl()))
		if json != nil {
			return json
		}
	}
	return wrapper.NewSVObject()
}

func (this *InnerEx) String() string {
	if len(this.subscribeTopic) > 0 {
		return fmt.Sprintf("InnerEx-%s", this.subscribeTopic)
	}

	if this.ownerPlug != nil {
		return fmt.Sprintf("%v(%s)-InnerEx", this.ownerPlug, reflect.TypeOf(this.ownerPlug))
	}
	return fmt.Sprintf("%p-InnerEx", this)
}

func (this *InnerEx) StatusString() string {
	if this.plug == nil {
		return fmt.Sprintf("%s(nil)", this.plugID)
	} else {
		return fmt.Sprintf("%s-%s(%p)-%s", this.plugID, reflect.TypeOf(this.plug), this.plug, this.task_topic)
	}
}

func (this *InnerEx) SetPlug(val interface{}) {
	this.plug = val
	if val == nil {
		return
	}
}

func (this *InnerEx) CheckDNA() uint32 {
	if this.objdna == 0 {
		this.objdna = utils.NewObjectDNA()
	}
	return this.objdna
}

func (this *InnerEx) unsubPre() {
	if len(this.subscribeTopic) > 0 {
		ex := this.checkGetPlug()
		if ex == nil {
			return
		}
		if sc, ok := ex.(utils.IUnsub); ok {
			sc.Unsub(this.subid, this.subscribeTopic)
		}
		this.subscribeTopic = ""
	}
}

func (this *InnerEx) UnSubscribePre() {
	if len(this.subscribeTopic) > 0 {
		ex := this.checkGetPlug()
		if ex == nil {
			return
		}

		if len(this.subid) > 0 {
			this.unsubPre()
		}

		if sc, ok := ex.(utils.IUnSubscribe); ok {
			sc.UnsubscribeChannel(this.subscribeTopic, this)
		} else if sc, ok := ex.(IGetSubscriber); ok {
			sc.GetSubscriber().UnsubscribeChannel(this.subscribeTopic, this)
		} else if sc, ok := ex.(utils.SubscribeCenter); ok {
			sc.UnsubscribeChannel(this.subscribeTopic, this)
		} else {
			return
		}
		this.subscribeTopic = ""
	}
}

func (this *InnerEx) AddListener(groupid, id interface{}, ls utils.Listener_CallBack) bool {
	ex := this.checkGetPlug()
	if ex == nil {
		return false
	}

	if l, ok := ex.(utils.IListenerOpera); ok {
		l.AddListener(groupid, id, ls)
		return true
	}
	return false

}

func (this *InnerEx) RemoveListener(groupid, id interface{}) {
	ex := this.checkGetPlug()
	if ex == nil {
		return
	}

	if l, ok := ex.(utils.IListenerOpera); ok {
		l.RemoveListener(groupid, id)
		return
	}
	return
}

func (this *InnerEx) SubscribeFromConf(conf *wrapper.SuperValue, onInput utils.SubscribeCallback) (err error) {
	if conf == nil {
		return nil
	}

	strTopic := conf.StringByName("topic", "")
	strTopic = utils.ReplacePlaceholder(strTopic, "$$", "$$")
	ReplacePlaceholderForFileName(strTopic)

	return this.Subscribe(strTopic, onInput)
}

func (this *InnerEx) SubscribeFromConfEx(conf *wrapper.SuperValue, onInput utils.SubFunc) (err error) {
	if conf == nil {
		return nil
	}

	strTopic := conf.StringByName("topic", "")
	strTopic = utils.ReplacePlaceholder(strTopic, "$$", "$$")
	ReplacePlaceholderForFileName(strTopic)

	return this.Sub(strTopic, onInput)
}

func (this *InnerEx) CheckGetRequestIntf() utils.IRequestIntf {
	if this.ReleaseFlag() == 1 {
		return nil
	}

	plg := this.checkGetPlug()
	if plg == nil {
		return nil
	}

	lvIntf, _ := plg.(utils.IRequestIntf)
	return lvIntf
}

func (this *InnerEx) CheckGetValueIntf() utils.IGetValue {
	if this.ReleaseFlag() == 1 {
		return nil
	}

	plg := this.checkGetPlug()
	if plg == nil {
		return nil
	}

	lvIntf, _ := plg.(utils.IGetValue)
	return lvIntf
}

func (this *InnerEx) TryGetIPub() (subpub.IPub, bool) {
	ex := this.checkGetPlug()
	if ex == nil {
		return nil, false
	}
	intf, ok := ex.(utils.IPub)
	return intf, ok
}

func (this *InnerEx) SupportIPub() bool {
	ex := this.checkGetPlug()
	if ex == nil {
		return false
	}
	if _, ok := ex.(utils.IPub); ok {
		return true
	} else {
		return false
	}
}

func (this *InnerEx) SupportISub() bool {
	ex := this.checkGetPlug()
	if ex == nil {
		return false
	}
	if _, ok := ex.(utils.ISub); ok {
		return true
	} else {
		return false
	}
}

// topic 插件如果有设定topic 优先使用
// 0:没有发送
// -1:发送失败
func (this *InnerEx) PubEx(topic string, max int, args ...interface{}) int {
	if this.ReleaseFlag() == 1 {
		return 0
	}

	ex := this.checkGetPlug()
	if ex == nil {
		return 0
	}

	if len(this.task_topic) > 0 {
		topic = this.task_topic
	}

	if len(this.topic_prefix) > 0 {
		topic = this.topic_prefix + topic
	}

	if intf, ok := ex.(utils.IPub); ok {
		return intf.Pub(topic, max, args...)
	} else if intf, ok := ex.(IDispatchData); ok {
		if len(args) >= 1 {
			err := intf.DispatchData(topic, this, args[0], nil)
			if err == nil {
				return 1
			} else {
				return -1
			}
		} else {
			return 0
		}
	} else {
		return 0
	}
}

/*
-1： 发送异常
*/
func (this *InnerEx) Pub(topic string, max int, args ...interface{}) int {
	if this.ReleaseFlag() == 1 {
		return -1
	}

	ex := this.checkGetPlug()
	if ex == nil {
		return -1
	}

	if len(topic) == 0 {
		topic = this.task_topic
	}

	if len(this.topic_prefix) > 0 {
		topic = this.topic_prefix + topic
	}

	if intf, ok := ex.(utils.IPub); ok {
		return intf.Pub(topic, max, args...)
	} else {
		return -1
	}
}

func (this *InnerEx) Unsub(id, topic string) bool {
	ex := this.checkGetPlug()
	if ex == nil {
		return false
	}
	if sc, ok := ex.(utils.IUnsub); ok {
		return sc.Unsub(id, topic)
	} else {
		return false
	}
}

func (this *InnerEx) SubEx(id, topic string, onfunc utils.SubFunc) (err error) {
	if this.ReleaseFlag() == 1 {
		return fmt.Errorf("当前插件已经被释放")
	}

	if len(this.topic_prefix) > 0 {
		topic = this.topic_prefix + topic
	}

	if topic == this.subscribeTopic {
		return
	}

	ex := this.checkGetPlug()
	if ex == nil {
		return fmt.Errorf("[%s]找不到对应的插件进行订阅", this.plugID)
	}
	if intf, ok := ex.(utils.ISub); ok {
		intf.Sub(id, topic, onfunc)
		return nil
	} else {
		return fmt.Errorf("不支持ISub接口")
	}
}

/*
会把之前的取消订阅
*/
func (this *InnerEx) Sub(topic string, onfunc utils.SubFunc) (err error) {
	if this.ReleaseFlag() == 1 {
		return fmt.Errorf("当前插件已经被释放")
	}

	if len(this.topic_prefix) > 0 {
		topic = this.topic_prefix + topic
	}

	if topic == this.subscribeTopic {
		return
	}

	ex := this.checkGetPlug()
	if ex == nil {
		return fmt.Errorf("[%s]找不到对应的插件进行订阅", this.plugID)
	}
	this.UnSubscribePre()

	this.subscribeTopic = topic
	if len(this.subscribeTopic) == 0 {
		return fmt.Errorf("没有设定订阅主题")
	}

	if intf, ok := ex.(utils.ISub); ok {
		subid := fmt.Sprintf("%s-%p", this.subscribeTopic, this)
		intf.Sub(subid, topic, onfunc)
		this.subid = subid
		return nil
	} else {
		return fmt.Errorf("不支持ISub接口")
	}
}

/*
**
可以重复执行, 如果topic和之前的不一致, 会先取消订阅之前的
*/
func (this *InnerEx) Subscribe(topic string, onInput utils.SubscribeCallback) (err error) {
	if this.ReleaseFlag() == 1 {
		return fmt.Errorf("当前插件已经被释放")
	}

	if len(this.topic_prefix) > 0 {
		topic = this.topic_prefix + topic
	}

	if topic == this.subscribeTopic {
		return
	}

	ex := this.checkGetPlug()
	if ex == nil {
		return fmt.Errorf("[%s]找不到对应的插件进行订阅", this.plugID)
	}

	this.UnSubscribePre()

	this.subscribeTopic = topic
	if len(this.subscribeTopic) == 0 {
		return fmt.Errorf("没有设定订阅主题")
	}

	if sc, ok := ex.(utils.ISubscribe); ok {
		sc.SubscribeChannel(topic, this, onInput)
	} else if sc, ok := ex.(IGetSubscriber); ok {
		sc.GetSubscriber().SubscribeChannel(topic, this, onInput)
	} else if sc, ok := ex.(utils.SubscribeCenter); ok {
		sc.SubscribeChannel(topic, this, onInput)
	}
	return nil
}

func (this *InnerEx) UnSubscribeEx(topic string, id interface{}) {
	ex := this.checkGetPlug()
	if ex == nil {
		return
	}
	if sc, ok := ex.(utils.IUnSubscribe); ok {
		sc.UnsubscribeChannel(topic, id)
	} else if sc, ok := ex.(IGetSubscriber); ok {
		sc.GetSubscriber().UnsubscribeChannel(topic, id)
	} else if sc, ok := ex.(utils.SubscribeCenter); ok {
		sc.UnsubscribeChannel(topic, id)
	} else {
		return
	}

}

/*
**

	可以重复执行, 不会取消之前的订阅, 需要自己执行取消
*/
func (this *InnerEx) SubscribeEx(topic string, id interface{}, onInput utils.SubscribeCallback) (subtopic string, err error) {
	if this.ReleaseFlag() == 1 {
		return "", fmt.Errorf("当前插件已经被释放")
	}

	if len(topic) == 0 {
		return "", fmt.Errorf("请指定订阅主题!")
	}

	if len(this.topic_prefix) > 0 {
		topic = this.topic_prefix + topic
	}

	ex := this.checkGetPlug()
	if ex == nil {
		return "", fmt.Errorf("[%s]找不到对应的插件进行订阅", this.plugID)
	}

	if sc, ok := ex.(utils.ISubscribe); ok {
		sc.SubscribeChannel(topic, id, onInput)
	} else if sc, ok := ex.(IGetSubscriber); ok {
		sc.GetSubscriber().SubscribeChannel(topic, id, onInput)
	} else if sc, ok := ex.(utils.SubscribeCenter); ok {
		sc.SubscribeChannel(topic, id, onInput)
	} else {
		return "", fmt.Errorf("[%s]插件[%v]不能进行阅", this.plugID, this.plug)
	}
	return topic, nil
}

/*
*

	{
	    "interval":10,
	    "topic":"xxxx"
	}
*/
func (this *InnerEx) ReloadTaskConf(conf *wrapper.SuperValue,
	status_fn func(), status_fn_start func(), status_fn_shutdown func()) (err error) {

	if conf == nil {
		return
	}

	conf.CheckStrFunc("topic", func(strv string) {
		this.task_topic = strv
	})

	conf.CheckExistsIntFunc("interval", func(val int64) {
		if this.task_interval != int(val) {
			this.task_intervalDNA++
			if this.task_interval <= 0 { // 之前是关闭状态
				if status_fn_start != nil {
					status_fn_start()
				}
			}
			this.task_interval = int(val)
			if this.task_interval > 0 {
				dna := this.task_intervalDNA
				this.CheckDNA()
				utils.AddTaskFxiedInterval(fmt.Sprintf("obj_%d", this.objdna),
					time.Duration(this.task_interval)*time.Second, func() bool {
						if dna != this.task_intervalDNA || this.ReleaseFlag() == 1 {
							if status_fn_shutdown != nil {
								status_fn_shutdown()
							}
							return false
						}
						if status_fn != nil {
							status_fn()
						}
						return true
					})
			} else {
				if status_fn_shutdown != nil {
					status_fn_shutdown()
				}
			}

		}

	})
	return
}

func (this *InnerEx) ReloadConfAndSubscribe(conf *wrapper.SuperValue, fn utils.SubscribeCallback) (err error) {
	e1 := this.ReloadConf(conf)
	if e1 != nil {
		return e1
	}

	return this.SubscribeFromConf(conf, fn)
}

func (this *InnerEx) checkClosePubQue() {
	if this.pubqueTask != nil {
		this.pubqueTask = nil
		chantask.DefaultChanTaskHub.ReleaseChanTask(this.pubque)
	}
	this.pubfun = this.innerPubFunc
}

func (this *InnerEx) GetOutputTopic() string {
	return this.task_topic
}

/*
	{
	    "id":"ex",
	    "jsonfields":"key=orgkey;orgkey2",
	    "templ":{}
	}
*/
func (this *InnerEx) ReloadConf(conf *wrapper.SuperValue) (err error) {
	this.confIsChanged = false
	if conf == nil {
		this.Reset()
		return
	}

	if conf.IsObject() {
		conf.CheckStrFunc("id", func(strv string) {
			if this.plugID != strv {
				this.UnSubscribePre()
				this.plugID = strv
				ex := FindPluginByID(strv)
				this.SetPlug(ex)
				this.confIsChanged = true
			}
		})

		conf.CheckExistsStrFunc("templ", func(strv string) {
			strv = utils.ReplacePlaceholder(strv, "$$", "$$")
			if this.templ != strv {
				this.templ = strv
				this.confIsChanged = true
			}
		})

		conf.CheckExistsStrFunc("jsonfields", func(strv string) {
			this.jsonfields = strv
		})

		conf.CheckExistsStrFunc("topic", func(strv string) {
			strv = utils.ReplacePlaceholder(strv, "$$", "$$")
			this.task_topic = strv
		})

		conf.CheckExistsStrFunc("topic-prefix", func(strv string) {
			this.topic_prefix = strv
		})

		pubque := conf.StringByName("pubque", "")
		if this.pubque != pubque {
			this.checkClosePubQue()
		}
		if len(this.pubque) > 0 {
			this.pubqueTask, _ = chantask.DefaultChanTaskHub.CheckGetChanTask(this.pubque, 1024, 0)
			this.pubfun = this.innerPubBufQueue
		} else {
			this.pubfun = this.innerPubFunc
		}
	} else {
		if this.plugID != conf.String() {
			this.pubfun = this.innerPubFunc
			this.UnSubscribePre()
			this.plugID = conf.String()
			ex := FindPluginByID(this.plugID)
			this.SetPlug(ex)
			this.confIsChanged = true
		}
	}

	return
}

func (this *InnerEx) checkGetPlug() interface{} {
	ex := this.plug
	if ex == nil {
		if len(this.plugID) == 0 {
			return nil
		}
		ex = FindPluginByID(this.plugID)
		if ex != nil {
			this.plug = ex
		}

	}
	return ex
}

func (this *InnerEx) CheckGetPlug() interface{} {
	return this.checkGetPlug()
}

func (this *InnerEx) OutputBufferTag(buffer []byte, tag interface{}) {
	if this.ReleaseFlag() == 1 {
		return
	}

	ex := this.checkGetPlug()
	if ex == nil {
		return
	}
	if sc, ok := ex.(IDispatchData); ok {
		sc.DispatchData(this.task_topic, this, buffer, tag)
	} else if sc, ok := ex.(IGetSubscriber); ok {
		sc.GetSubscriber().DispatchData(this.task_topic, this, buffer, tag)
	} else if sc, ok := ex.(utils.SubscribeCenter); ok {
		sc.DispatchData(this.task_topic, this, buffer, tag)
	} else if sc, ok := ex.(ISendKVs); ok {
		sc.SendKVs(this.task_topic, buffer)
	}
}

/*
*

	注意内部数据是异步还是同步，如果是异步，数据需要进行copy
*/
func (this *InnerEx) OutputTaskBuffer(buffer []byte) {
	if this.ReleaseFlag() == 1 {
		return
	}
	this.OutputBufferEx(buffer, nil, nil)
}

func (this *InnerEx) OutputBufferEx(buffer []byte, sender interface{}, tag interface{}) {
	if this.ReleaseFlag() == 1 {
		return
	}

	ex := this.checkGetPlug()
	if ex == nil {
		return
	}
	if sc, ok := ex.(IDispatchData); ok {
		sc.DispatchData(this.task_topic, sender, buffer, tag)
	} else if sc, ok := ex.(IGetSubscriber); ok {
		sc.GetSubscriber().DispatchData(this.task_topic, sender, buffer, tag)
	} else if sc, ok := ex.(utils.SubscribeCenter); ok {
		sc.DispatchData(this.task_topic, sender, buffer, tag)
	} else if sc, ok := ex.(ISendKVs); ok {
		sc.SendKVs(this.task_topic, buffer)
	} else {
		golog.Warnf("插件[%s]不支持IDispatchData/ISendKVs/IGetSubscriber/SubscribeCenter", this.plugID)
	}
}

func (this *InnerEx) SendTopicBuffer(topic string, buffer []byte) {
	if this.ReleaseFlag() == 1 {
		return
	}

	if len(topic) == 0 {
		topic = this.task_topic
	}

	if len(this.topic_prefix) > 0 {
		topic = this.topic_prefix + topic
	}

	ex := this.checkGetPlug()
	if ex == nil {
		return
	}
	if sc, ok := ex.(IDispatchData); ok {
		sc.DispatchData(topic, this, buffer, nil)
	} else if sc, ok := ex.(ISendKVs); ok {
		sc.SendKVs(topic, buffer)
	} else if sc, ok := ex.(IGetSubscriber); ok {
		sc.GetSubscriber().DispatchData(topic, nil, buffer, nil)
	} else if sc, ok := ex.(utils.SubscribeCenter); ok {
		sc.DispatchData(topic, nil, buffer, nil)
	} else {
		golog.Warnf("插件[%s]不支持IDispatchData/ISendKVs/IGetSubscriber/SubscribeCenter", this.plugID)
	}
}

func (this *InnerEx) RequestSV(req *wrapper.SuperValue, resp *wrapper.SuperValue) {
	if this.ReleaseFlag() == 1 {
		if resp != nil {
			resp.SetKeyValues("code", -1, "msg", "ReleaseFlag==1, 无法进行调用RequestSV")
		}
		return
	}
	ex := this.checkGetPlug()
	if ex == nil {
		if resp != nil {
			resp.SetKeyValues("code", -1, "msg", fmt.Sprintf("%s is nil, 无法进行调用RequestSV", this.plugID))
		}
		return
	}
	if intf, ok := ex.(wrapper.ISVRequest); ok {
		intf.RequestSV(req, resp)
	} else {
		if resp != nil {
			resp.SetKeyValues("code", -1, "msg", fmt.Sprintf("%s 不支持ISVRequest, 无法进行调用RequestSV", this.plugID))
		}
	}
}

func (this *InnerEx) DispatchData(topic string, sender interface{}, data interface{}, tag interface{}) {
	if this.ReleaseFlag() == 1 {
		utils.TryCallBack(data, sender, fmt.Errorf("插件已经释放!"))
		return
	}
	ex := this.checkGetPlug()
	if ex == nil {
		utils.TryCallBack(data, sender, fmt.Errorf("找不到对应的插件(%s)", this.plugID))
		return
	}

	//if len(this.templ) > 0 {
	//	if json, ok := data.(*wrapper.SuperValue); ok {
	//		jsonTempl := this.createReportJSON();
	//		json.CopyFromRef(jsonTempl)
	//	}
	//}

	if len(topic) == 0 {
		topic = this.task_topic
	}

	if len(this.topic_prefix) > 0 {
		topic = this.topic_prefix + topic
	}

	if sc, ok := ex.(IDispatchData); ok {
		// 内部去调用TryCallBack
		sc.DispatchData(topic, sender, data, tag)
		//err := sc.DispatchData(topic, sender, data, tag)
		//if err != nil {
		//	utils.TryCallBack(data, sender, err)
		//	golog.Warnf("[%s]分发数据出现异常:%s", this.TypeStr, err.Error())
		//}
	} else if sc, ok := ex.(ISendKVs); ok {
		sc.SendKVs(topic, data)
		//err := sc.SendKVs(topic, data)
		//if err != nil {
		//	utils.TryCallBack(data, sender, err)
		//	golog.Warnf("[%s]分发数据出现异常:%s", this.TypeStr, err.Error())
		//}
	} else if sc, ok := ex.(IGetSubscriber); ok {
		sc.GetSubscriber().DispatchData(topic, sender, data, tag)
	} else if sc, ok := ex.(utils.SubscribeCenter); ok {
		sc.DispatchData(topic, sender, data, tag)
	} else {
		utils.TryCallBack(data, sender, fmt.Errorf("插件[%s]不支持IDispatchData/ISendKVs/IGetSubscriber/SubscribeCenter", this.plugID))
		golog.Warnf("插件[%s]不支持IDispatchData/ISendKVs/IGetSubscriber/SubscribeCenter", this.plugID)
	}
}

func (this *InnerEx) CheckValid() bool {
	if this.ReleaseFlag() == 1 {
		return false
	}
	return this.checkGetPlug() != nil

}

func (this *InnerEx) SendJSON(topic string, json *wrapper.SuperValue) {
	if this.ReleaseFlag() == 1 {
		return
	}
	this.DispatchData(topic, this, json, nil)
}

func (this *InnerEx) innerPubBufQueue(topic string, args ...interface{}) int {
	pubtask := this.pubqueTask
	if pubtask != nil {
		pubtask.PushTask(func(args ...interface{}) {
			if atomic.LoadInt32(&this.releaseFlag) == 1 {
				return
			}
			this.innerPubFunc(topic, args...)
		}, args...)
		return 1
	} else {
		return 0
	}
}

func (this *InnerEx) innerPubFunc(topic string, args ...interface{}) int {
	if intf, ok := this.TryGetIPub(); ok {
		return intf.Pub(topic, 0, args...)
	} else {
		return -1
	}
}

func (this *InnerEx) PubJSON(topic string, fn func(json *wrapper.SuperValue)) error {
	if this.ReleaseFlag() == 1 {
		return io.EOF
	}

	json := this.CreateReportJSONCache(true)
	defer json.Clear()
	fn(json)
	pubfn := this.pubfun
	data := json.AsJSONStringEx(false, false)
	pubfn(topic, data)
	return nil
}

func (this *InnerEx) SendTaskTopic(fillStatus func(json *wrapper.SuperValue)) {
	if this.ReleaseFlag() == 1 {
		return
	}
	this.SendTopic(this.task_topic, fillStatus)

}

func (this *InnerEx) SendTopicEx(topic string, fillStatus func(json *wrapper.SuperValue)) {
	if len(this.topic_prefix) > 0 {
		topic = this.topic_prefix + topic
	}
	this.SendTopic(topic, fillStatus)
}

func (this *InnerEx) SendTopic(topic string, fillStatus func(json *wrapper.SuperValue)) {
	if this.ReleaseFlag() == 1 {
		return
	}

	ex := this.checkGetPlug()
	if ex == nil {
		return
	}

	if sc, ok := ex.(IDispatchData); ok {
		json := this.createReportJSON()
		fillStatus(json)
		if len(this.jsonfields) > 0 {
			newjson := wrapper.NewSVObject()
			newjson.CopyFromByFields(json, this.jsonfields)
			json = newjson
		}
		err := sc.DispatchData(topic, this, json, nil)
		if err != nil {
			golog.Warnf("[%s]分发数据出现异常:%s", this.TypeStr, err.Error())
		}
	} else if sc, ok := ex.(ISendKVs); ok {
		json := this.createReportJSON()
		fillStatus(json)
		if len(this.jsonfields) > 0 {
			newjson := wrapper.NewSVObject()
			newjson.CopyFromByFields(json, this.jsonfields)
			json = newjson
		}
		err := sc.SendKVs(topic, json)
		if err != nil {
			golog.Warnf("[%s]分发数据出现异常:%s", this.TypeStr, err.Error())
		}
	} else if sc, ok := ex.(IGetSubscriber); ok {
		json := this.createReportJSON()
		fillStatus(json)
		if len(this.jsonfields) > 0 {
			newjson := wrapper.NewSVObject()
			newjson.CopyFromByFields(json, this.jsonfields)
			json = newjson
		}
		sc.GetSubscriber().DispatchData(topic, this, json, nil)
	} else if sc, ok := ex.(utils.SubscribeCenter); ok {
		json := this.createReportJSON()
		fillStatus(json)
		if len(this.jsonfields) > 0 {
			newjson := wrapper.NewSVObject()
			newjson.CopyFromByFields(json, this.jsonfields)
			json = newjson
		}
		sc.DispatchData(topic, this, json, nil)
	} else {
		golog.Warnf("插件[%s]不支持IDispatchData/ISendKVs/IGetSubscriber/SubscribeCenter", this.plugID)
	}
}

func (this *InnerEx) SetValue(key string, value interface{}) error {
	if this.ReleaseFlag() == 1 {
		return fmt.Errorf("插件已经释放!")
	}

	plg := this.checkGetPlug()
	if plg == nil {
		return fmt.Errorf("插件不存在")
	}

	intf, ok := plg.(IPlgMap)
	if !ok {
		return fmt.Errorf("插件[%s]不支持IPlgMap接口", this.plugID)
	}
	return intf.SetValue(key, value)
}

func (this *InnerEx) GetValue(key string) (val interface{}, err error) {
	if this.ReleaseFlag() == 1 {
		return nil, fmt.Errorf("插件已经释放!")
	}

	plg := this.checkGetPlug()
	if plg == nil {
		return nil, fmt.Errorf("插件不存在")
	}

	intf, ok := plg.(IPlgMap)
	if !ok {
		return nil, fmt.Errorf("插件[%s]不支持IPlgMap接口", this.plugID)
	}
	return intf.GetValue(key)
}

func (this *InnerEx) RemoveValue(key string) (ok bool, err error) {
	if this.ReleaseFlag() == 1 {
		return false, fmt.Errorf("插件已经释放!")
	}

	plg := this.checkGetPlug()
	if plg == nil {
		return false, fmt.Errorf("插件不存在")
	}

	intf, ok := plg.(IPlgMap)
	if !ok {
		return false, fmt.Errorf("插件[%s]不支持IPlgMap接口", this.plugID)
	}
	return intf.RemoveValue(key)
}
